#ifndef __IOTHREAD_H__
#define __IOTHREAD_H__

#include <pthread.h>
#include <unistd.h>
#include <cstdlib>
#include <atomic>
#include "server.h"

#define IO_THREAD_MAX_NUM 128
#define IO_THREAD_OP_READ 0
#define IO_THREAD_OP_WRITE 1

static pthread_t tid_[IO_THREAD_MAX_NUM];
static pthread_mutex_t mtx_[IO_THREAD_MAX_NUM];
static std::atomic<unsigned long> pending_[IO_THREAD_MAX_NUM];//i位置待处理任务个数
static std::list<eventLoop*> tlist_[IO_THREAD_MAX_NUM];//不同线程需要处理的事件，从全局队列中取出放入
static int thread_op_ = IO_THREAD_OP_READ;//

int freeClients()
{
  std::list<int>::iterator it = s.close_fd_list_.begin();
  for(; it !=  s.close_fd_list_.end();)
  {
    //删除事件
    eventLoop* el = s.rt_->getEventLoop(*it);
    s.rt_->delEpollEvent(*it, el->mask_);


    close(*it);
     //std::cout << "close fd: " << *it << std::endl;
     s.close_fd_list_.erase(it++);
  }
  return 0;
}

static int getIOPendingCount(int i)
{
  if(i >= IO_THREAD_MAX_NUM)
    return 0;

  return pending_[i];
}

static void setIOPendingCount(int i, int count)
{
  if(i >= IO_THREAD_MAX_NUM)
    return;

  pending_[i] = count;
}

int handleClientReadUsingThread()
{
  int ret = s.client_read_list_.size();
  if(ret == 0)
    return 0;

  //插入到线程需要处理的数组中
  int item_id = 0;
  std::list<eventLoop*>::iterator it = s.client_read_list_.begin();
  for(; it != s.client_read_list_.end(); ++it)
  {
    int target_id = item_id % s.threadnum_;
    tlist_[target_id].push_back(*it);
    item_id++;
  }
  
  //设置任务个数
  thread_op_ = IO_THREAD_OP_READ;
  for(int i = 0; i < s.threadnum_; ++i)
  {
    int count = tlist_[i].size();
    setIOPendingCount(i, count);
  }

  //主线程也作为io线程处理任务
  std::list<eventLoop*>::iterator it2 = tlist_[0].begin();
  for(; it2 != tlist_[0].end(); ++it2)
  {
    readQueryFromClient(*it2);
  }
  tlist_[0].clear();

  //等待其它线程完成工作
  while(1)
  {
    unsigned long pending = 0;
    for(int i = 1; i < s.threadnum_; ++i)
    {
      pending += getIOPendingCount(i);
    }
    if(!pending)
      break;
  }
  
  //删除全局队列任务
  s.client_read_list_.clear();
  return ret;
}

void startThreadIO()
{
  if(s.io_thread_active_ == 1)
    return ;
  for(int i = 1; i < s.threadnum_; ++i)
  {
    pthread_mutex_unlock(&mtx_[i]);
  }
  s.io_thread_active_ = 1;
}

void stopThreadIO()
{
  //停止前读一下数据
  handleClientReadUsingThread();
  if(s.io_thread_active_ == 0)
    return;

  for(int i = 1; i < s.threadnum_; ++i)
  {
    pthread_mutex_lock(&mtx_[i]);
  }
  s.io_thread_active_ = 0;
  return ;
}

int stopThreadIOIfNeed()
{
  return 0;
  
  int num = s.client_write_list_.size();
  if(num == 1)
    return 1;

  if(num < (s.threadnum_ * 2))
  {
    if(s.io_thread_active_)
      stopThreadIO();
    return 1;
  }
  return 0;
}

int handleClientWrite()
{
  int ret = s.client_write_list_.size();
  //直接发送
  std::list<eventLoop*>::iterator it = s.client_write_list_.begin();
  for(; it != s.client_write_list_.end(); )
  {
    (*it)->flag_ &= ~CLIENT_PENDING_READ;
    writeToClient(*it);
    s.client_write_list_.erase(it++);
  }
  return ret;
}

int handleClientWriteUsingThread()
{
  int ret = s.client_write_list_.size();
  if(ret == 0)
    return 0;

  if(s.threadnum_ == 1 || stopThreadIOIfNeed())
    return handleClientWrite();

  if(!s.io_thread_active_)
    startThreadIO();


  //插入到线程需要处理的数组中
  int item_id = 0;
  std::list<eventLoop*>::iterator it = s.client_write_list_.begin();
  for(; it != s.client_write_list_.end(); ++it)
  {
    int target_id = item_id % s.threadnum_;
    tlist_[target_id].push_back(*it);
    item_id++;
  }
  
  //设置任务个数
  thread_op_ = IO_THREAD_OP_WRITE;
  for(int i = 0; i < s.threadnum_; ++i)
  {
    int count = tlist_[i].size();
    setIOPendingCount(i, count);
  }

  //主线程也作为io线程处理任务
  std::list<eventLoop*>::iterator it2 = tlist_[0].begin();
  for(; it2 != tlist_[0].end(); ++it2)
  {
    writeToClient(*it2);
  }
  tlist_[0].clear();

  //等待其它线程完成工作
  while(1)
  {
    unsigned long pending = 0;
    for(int i = 1; i < s.threadnum_; ++i)
    {
      pending += getIOPendingCount(i);
    }
    if(!pending)
      break;
  }
  
  //删除全局队列任务
  s.client_write_list_.clear();
  return ret;
}


static void* IOThreadMain(void* arg)
{
  long idx = (unsigned long)arg;

  while(1)
  {
    if(getIOPendingCount(idx) == 0)
    {
      pthread_mutex_lock(&mtx_[idx]); 
      pthread_mutex_unlock(&mtx_[idx]);
      continue;
    }

    std::list<eventLoop*>::iterator e_it = tlist_[idx].begin();
    for(; e_it != tlist_[idx].end(); ++e_it)
    {
      std::cout << "IOThreadMain...." << pthread_self() << std::endl;
      eventLoop* el = *e_it;
      if(thread_op_ == IO_THREAD_OP_READ)
        readQueryFromClient(el);
      if(thread_op_ ==  IO_THREAD_OP_WRITE)
        writeToClient(el);
      // if(el->mask_ && AE_READABLE)
      //   readQueryFromClient(el);
      // if(el->mask_ && AE_WRITEABLE)
      //   writeToClient(el);
    }
    tlist_[idx].clear();
    setIOPendingCount(idx, 0);
  }
  
  return (void*)0;
}

void initThreadIOPool()
{
  memset(tid_, 0, IO_THREAD_MAX_NUM); 
  memset(mtx_, 0, IO_THREAD_MAX_NUM); 
  
  if(s.threadnum_ <= 1)
    return;

  for(int i = 1; i < s.threadnum_; ++i)
  {
    pthread_mutex_init(&mtx_[i], nullptr);
    std::cout << pthread_self() << "lock..." << std::endl;
    pthread_mutex_lock(&mtx_[i]);

    pthread_t tid = 0;
    if(pthread_create(&tid, nullptr, IOThreadMain, (void*)(long)i) < 0)
    {
      exit(-1);
    }
    std::cout << "initThreadIOPool create pthread id: " << tid << std::endl;
    tid_[i] = tid;
  }
}

#endif
